package com.hcy.rabbitruntimeapi.websocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hcy.rabbitruntimeapi.handler.MyWebSocketFrameHandler;
import com.hcy.rabbitruntimeapi.model.dto.WebSocketKey;
import com.hcy.rabbitbasicmodel.common.MyWebSocketMessage;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.nio.charset.CharsetDecoder;
import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;

import static javax.swing.UIManager.get;

/**
 * @Author huchenying
 * @Description 推送websocekt消息
 * @Date 2022/3/19 20:49
 */
@Component
public class WebSocketMQConsumer implements ApplicationRunner {
    private static final Logger logger = LoggerFactory.getLogger(WebSocketMQConsumer.class);

    @Value("${MQ.namesrvAddr}")
    private String addr;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("WEBSOCKET");
        //设置NameServer地址
        consumer.setNamesrvAddr(addr);
        //设置起始消费位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //指定消费Topic和Tag
        consumer.subscribe("WEBSOCKET","*");
        //指定消费模式 默认集群模式
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setConsumeThreadMin(4);
        consumer.setConsumeThreadMax(8);
        consumer.setConsumeMessageBatchMaxSize(16);
        //注册消息监听器
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            //逐条消费消息
            for (MessageExt msg : msgs) {
                try {
                    String mmsg = new String(msg.getBody());
                    MyWebSocketMessage webSocketMessage = JSON.parseObject(mmsg, MyWebSocketMessage.class);
                    consumer(webSocketMessage,msg.getMsgId());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            //返回消费状态
            //重新消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //开启消费
        consumer.start();
        logger.info("websocket mq consumer started");
    }

    public void consumer(MyWebSocketMessage message,String msg){
        logger.info("consumer start param -> message =  {},clientId = {}", ToStringBuilder.reflectionToString(message),msg);
        WebSocketKey webSocketKey = new WebSocketKey(message.getUsername(),message.getClientId());
        Channel channel = MyWebSocketFrameHandler.channels.get(webSocketKey);
        if(channel == null){ // 当前用户不在线
            logger.info("当前用户不在线 webSocketKey -> {}",ToStringBuilder.reflectionToString(webSocketKey));
            //持久化消息，等用户上线拉取
        }else {
            try {
                channel.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(message)));
            } catch (Exception e) {
                logger.error("消息发送失败 -> {}",ToStringBuilder.reflectionToString(e));
            }
        }
    }
}
